﻿using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Channels;
using System.Threading.Tasks;

namespace SJ.Platform.Cache.RabbitMQ
{
    public class RabbitMQConnection : IRabbitMQConnection
    {
        private static RabbitMQConfig _rabbitMqConfig { get; set; }

        /*
        备注：使用数组的部分，是给消费端用的。目前生产者只设置了一个，消费者可能存在多个。
                    当然，有条件的还可以上RabbitMQ集群进行处理，会更好玩一点。
        */
        private static IConnectionFactory _producerFactory;  //RabbitMQ工厂 生产端
        private static IConnectionFactory[] _consumerFactory; //RabbitMQ工厂 消费端  

        private static IConnection _producer; //连接 生产端
        private static IConnection[] _consumer; //连接 消费端

        private static IModel _modelProducer;  //通道  生产端
        private static IModel[] _modelConsumer; //通道  消费端

        private static EventingBasicConsumer[] _basicConsumer;  // 事件

        /// <summary>
        /// 构造函数
        /// </summary>
        /// <param name="rabbitMqConfig">rabbitmq配置</param>
        /// <param name="msgFunc">消息处理委托</param>
        /// <param name="next">消息处理结果返回委托</param>
        public static void InitRabbitMQConnection(RabbitMQConfig rabbitMqConfig, Func<string, Task<bool>> msgFunc, Action<bool> next)
        {
            _rabbitMqConfig = rabbitMqConfig;

            //生产端
            ProducerConnection();

            //消费端 
            InitConsumer(msgFunc, next);
        }
        /// <summary>
        /// 生产者
        /// </summary>
        private static void ProducerConnection()
        {
            if (_producerFactory == null)
            {
                _producerFactory = new ConnectionFactory
                {
                    HostName = _rabbitMqConfig.Host,
                    Port = _rabbitMqConfig.Port,
                    UserName = _rabbitMqConfig.User,
                    Password = _rabbitMqConfig.Password
                };
            }

            if (_producer != null && _producer.IsOpen)
            {
                return;
            }
            _producer = _producerFactory.CreateConnection();
            if (_modelProducer != null && _modelProducer.IsOpen)
            {
                return;
            }
            _modelProducer = _producer.CreateModel();
            _modelProducer.ExchangeDeclare(_rabbitMqConfig.ExchangeName, _rabbitMqConfig.Direct);
        }

        public static void InitConsumer(Func<string, Task<bool>> msgFunc, Action<bool> next)
        {
            var reciveConn = new ConnectionFactory
            {
                HostName = _rabbitMqConfig.Host,
                Port = _rabbitMqConfig.Port,
                UserName = _rabbitMqConfig.User,
                Password = _rabbitMqConfig.Password
            };
            _consumer = new IConnection[_rabbitMqConfig.CostomerCount];
            _modelConsumer = new IModel[_rabbitMqConfig.CostomerCount];
            _basicConsumer = new EventingBasicConsumer[_rabbitMqConfig.CostomerCount];
            _consumerFactory = new IConnectionFactory[_rabbitMqConfig.CostomerCount];
            for (int i = 0; i < _rabbitMqConfig.CostomerCount; i++)
            {
                _consumerFactory[i] = reciveConn;

                if (null != _consumer[i] && _consumer[i].IsOpen)
                {
                    return;
                }
                _consumer[i] = _consumerFactory[i].CreateConnection(); // 创建消费者连接

                if (null != _modelConsumer[i] && _modelConsumer[i].IsOpen)
                {
                    return;
                }
                _modelConsumer[i] = _consumer[i].CreateModel();  // 创建消费者通道

                _basicConsumer[i] = new EventingBasicConsumer(_modelConsumer[i]);

                _modelConsumer[i].ExchangeDeclare(_rabbitMqConfig.ExchangeName, _rabbitMqConfig.Direct); // 定义交换机名称和类型  与生产者保持一致

                _modelConsumer[i].QueueDeclare(
                             queue: _rabbitMqConfig.QueueName, //消息队列名称
                             durable: _rabbitMqConfig.Durable, // 是否可持久化，此处配置在文件中，默认全局持久化(true)，也可以自定义更改
                             exclusive: false,
                             autoDelete: false,
                             arguments: null
               );  // 定义消费者队列

                _modelConsumer[i].QueueBind(_rabbitMqConfig.QueueName, _rabbitMqConfig.ExchangeName, _rabbitMqConfig.RoutingKey); // 队列绑定给指定的交换机

                _modelConsumer[i].BasicQos(0, 1, false); // 设置消费者每次只接收一条消息

                ConsumerConnection(i, msgFunc, next);
            }
        }

        /// <summary>
        /// 消费者处理消息
        /// </summary>
        /// <param name="consumeIndex">消费者序号</param>
        /// <param name="msgFunc">消息处理委托</param>
        private static void ConsumerConnection(int consumeIndex, Func<string, Task<bool>> msgFunc, Action<bool> next)
        {
            // 设置返回消息的回调函数
            _modelConsumer[consumeIndex].BasicReturn += (sender, eventArgs) =>
            {
                var body = eventArgs.Body.ToArray();
                var result = Encoding.UTF8.GetString(body);
                // 这里可以根据实际需求处理返回的消息，例如记录日志或者通知生产者
                next?.Invoke(Convert.ToBoolean(result));
            };

            StartListener((model, ea) =>
            {
                byte[] message = ea.Body.ToArray(); // 接收到的消息
                string msg = Encoding.UTF8.GetString(message);

                bool result = true;
                try
                {
                    result = msgFunc.Invoke(msg).Result;//回调方法
                }
                catch (Exception ex)
                {
                    result = false;
                }
                if (result)
                {
                    _modelConsumer[consumeIndex].BasicAck(ea.DeliveryTag, false);//确认消息被消费

                    // 发布消息到一个不存在的队列，以触发返回消息
                    _modelConsumer[consumeIndex].BasicPublish(
                        exchange: "",
                        routingKey: _rabbitMqConfig.QueueName + "_Result",
                        mandatory: true,
                        basicProperties: null,
                        body: Encoding.UTF8.GetBytes(result.ToString()));
                }
                else
                {
                    // 拒绝消息并将其重新放回队列
                    //_modelConsumer[consumeIndex].BasicNack(deliveryTag: ea.DeliveryTag, multiple: false, requeue: true);

                    // 拒绝消息并将其移到死信队列
                    _modelConsumer[consumeIndex].BasicReject(deliveryTag: ea.DeliveryTag, requeue: false);
                }
            }, _rabbitMqConfig.QueueName, consumeIndex);
        }

        /// <summary>
        /// 消费者接收消息的确认机制
        /// </summary>
        /// <param name="basicDeliverEventArgs"></param>
        /// <param name="queueName"></param>
        /// <param name="consumeIndex"></param>
        private static void StartListener(EventHandler<BasicDeliverEventArgs> basicDeliverEventArgs, string queueName, int consumeIndex)
        {
            _basicConsumer[consumeIndex].Received += basicDeliverEventArgs;
            //消费消息
            _modelConsumer[consumeIndex].BasicConsume(queue: queueName, autoAck: false, consumer: _basicConsumer[consumeIndex]); // 设置手动确认。
        }

        /// <summary>
        /// 消息发布
        /// </summary>
        /// <param name="message"></param>
        /// <param name="exchangeName"></param>
        /// <param name="routingKey"></param>
        public static void PublishExchange(string message, string exchangeName, string routingKey = "")
        {
            byte[] body = Encoding.UTF8.GetBytes(message);
            _modelProducer.BasicPublish(exchangeName, routingKey, null, body);
        }
    }
}